package com.niit.reduceJoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * Date:2024/11/12
 * Author：Ys
 * Description:
 */
public class ReduceJoinMapper extends Mapper<LongWritable, Text,Text,Text> {

    Text outKey = new Text();//输出的Key是一个文本 goodsId商品iD
    Text outValue = new Text();//输出的Value是一个文本   商品信息 或 订单信息

    StringBuilder sb = new StringBuilder();//是String的一个升级版的工具类，可以拼接字符串，比String类效率更高

    String fileName = null ; //文件名

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //获取当前处理切片所属的文件名字
        FileSplit inputSplit = (FileSplit)context.getInputSplit();
        fileName = inputSplit.getPath().getName();//通过context获得输入分片后，就可以根据路径下名字获取正在读取文件名称
        System.out.println("当前正在处理的文件："+ fileName);
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        //设置字符串的长度，此处用于清空数据
        sb.setLength(0); //为了防止数据重复，每次处理完数据后，需要清空数据，否则数据会重复

        //切割处理输入数据
        String[] fields = value.toString().split("\\|");// \是转义符，需要转义，否则会报错

        //判断处理的是哪个文件
        if(fileName.contains("my_goods.txt")){//处理的商品数据
            //100101|155083444927602|四川果冻橙6个约180g （商品id、商品编号、商品名称）
            String goodsId = fields[0]; //商品id
            outKey.set(goodsId);//设置输出的key
            String goodsNo = fields[1];//商品编号
            String goodsName = fields[2];//商品名称

            StringBuilder append = sb.append(goodsNo).append("\t").append(goodsName);
                                        //155083444927602 四川果冻橙6个约180g
            //在上述字符串的起始位置插入 goods#，目的是让reduce阶段分出mapper阶段输出的商品信息还是订单信息
            outValue.set(sb.insert(0,"goods#").toString());
        }else{//处理的订单数据
            // 2|100101|11192  （订单编号、商品id、实际支付价格）
            outKey.set(fields[1]);//设置输出的key 为商品ID
            sb.append(fields[0]).append("\t").append(fields[2]);//2|11192
            outValue.set(sb.insert(0,"order#").toString());//在起始位置插入order#

        }
        context.write(outKey,outValue);

    }
}
